package com.evil.common.rocketmq.interseptor;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.lang.UUID;
import com.alibaba.fastjson.JSON;
import com.evil.common.core.util.ServletUtils;
import com.evil.common.redis.util.RedisUtil;
import com.evil.common.rocketmq.annotation.MqCache;
import com.evil.common.rocketmq.entity.MsgCache;
import com.evil.common.rocketmq.util.MqRedisKeyUtil;
import com.evil.common.rocketmq.util.RocketMQUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.aspectj.lang.JoinPoint;
import org.aspectj.lang.annotation.AfterReturning;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Before;
import org.aspectj.lang.annotation.Pointcut;
import org.springframework.stereotype.Component;

import java.util.Set;

/**
 * MQ消息延迟发送 缓存
 *
 * @author liyang
 * @date 2021/11/11 10:41
 */
@Slf4j
@Aspect
@Component
@RequiredArgsConstructor
public class MqCacheAspect {

    private final RedisUtil redisUtil;
    private final RocketMQUtil rocketMQUtil;

    /**
     * 切面
     *
     * @param mqCache mqCache
     */
    @Pointcut("execution(public * *(..)) && @annotation(mqCache)")
    public void mqCachePointCut(MqCache mqCache) {
    }

    /**
     * 方法执行之前
     *
     * @param joinPoint joinPoint
     * @param mqCache   mqCache
     */
    @Before(value = "mqCachePointCut(mqCache)", argNames = "joinPoint,mqCache")
    public void before(JoinPoint joinPoint, MqCache mqCache) {
        // 获取签名对象
        String methodName = joinPoint.getSignature().getName();
        // 获取类名
        String className = joinPoint.getSignature().getDeclaringTypeName();

        String key = MqRedisKeyUtil.requestUuidMqMsgLevelKey();
        String level = UUID.fastUUID().toString(true);
        long l = redisUtil.sSet(key, level);
        log.info("mqCache before|class:{}, method:{}, level:{}, l:{}", className, methodName, level, l);
    }

    /**
     * 方法执行之后-正常返回
     *
     * @param joinPoint joinPoint
     * @param mqCache   mqCache
     */
    @AfterReturning(value = "mqCachePointCut(mqCache)", argNames = "joinPoint,mqCache")
    public void afterReturning(JoinPoint joinPoint, MqCache mqCache) {
        // 获取签名对象
        String methodName = joinPoint.getSignature().getName();
        // 获取类名
        String className = joinPoint.getSignature().getDeclaringTypeName();
        // 获取最新的元素-从左边push就从左边pop
        Set<Object> levels = redisUtil.sGet(MqRedisKeyUtil.requestUuidMqMsgLevelKey());
        log.info("mqCache before|class:{}, method:{}, level.size:{}", className, methodName, CollectionUtil.size(levels));
        if (CollectionUtil.isEmpty(levels)) {
            return;
        }
        int size = levels.size();
        String level = (String) levels.iterator().next();
        // 最后一层
        if (size == 1) {
            // 执行发送MQ
            String key = MqRedisKeyUtil.requestUuidMqMsgCacheKey();
            // 轮询缓存mq msg
            while (true) {
                MsgCache msgCache = (MsgCache) redisUtil.lpop(key);
                if (null == msgCache) {
                    break;
                }
                String uuid = msgCache.getUuid();
                String destination = msgCache.getDestination();
                Object msg = msgCache.getMsg();
                log.info("afterReturning|uuid:{}, destination:{}", uuid, destination);
                SendResult result = rocketMQUtil.syncSendWithHeaders(destination, msg, ServletUtils.getHeaderMap());
                log.info("afterReturning|uuid:{}, destination:{}, result:{}", uuid, destination, JSON.toJSONString(result));
            }
            return;
        }
        redisUtil.setRemove(level);
    }
}
